package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.Func.AsyncDimFunction;
import com.atguigu.Func.MyWindowFunction;
import com.atguigu.bean.ExamPaperScoreBean;
import com.atguigu.util.ClickHouseUtil_hjy;
import com.atguigu.util.MyKafkaUtil_hjy;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @author hjy
 * @create 2023/3/29 10:43
 */
public class Dws_examPaperScoreWindow {
    public static void main(String[] args) throws Exception {
        //todo 1 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://47.102.112.46:8020/gmall-flink/check/baseLogApp");
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
//        env.setStateBackend(new HashMapStateBackend());
//        System.setProperty("HADOOP_USER_NAME","atguigu");
        //todo 2 从kafka主题（dwd_test_exam_question）消费数据
        String topic="dwd_test_exam";
        String groupId="dws_exam_paper_score_window";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil_hjy.getKafkaConsumer(topic, groupId));
        //todo 3 将数据转为JSONObj 并过滤 并提取时间戳
        SingleOutputStreamOperator<JSONObject> jsonObjWithWM = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject record = JSON.parseObject(value);
                out.collect(record);
            }
        })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(TimestampAssignerSupplier.of(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("create_time");
                            }
                        })));
        //todo 4 按照paper_id分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjWithWM.keyBy(value -> value.getString("paper_id"));
        //todo 5 转为javaBean
        SingleOutputStreamOperator<ExamPaperScoreBean> javaBeanDS = keyedStream.map(new MapFunction<JSONObject, ExamPaperScoreBean>() {
            @Override
            public ExamPaperScoreBean map(JSONObject value) throws Exception {
                String paper_id = value.getString("paper_id");
                Double score = value.getDouble("score");
                Long A = 0L;
                Long B = 0L;
                Long C = 0L;
                Long D = 0L;
                if (score >= 90) {
                    A = 1L;
                } else if (score >= 75) {
                    B = 1L;
                } else if (score >= 60) {
                    C = 1L;
                } else {
                    D = 1L;
                }
                return ExamPaperScoreBean.builder()
                        .paper_id(paper_id)
                        .ACt(A)
                        .BCt(B)
                        .CCt(C)
                        .DCt(D)
                        .build();
            }
        });
        //todo 6 按照paper_id分组
        KeyedStream<ExamPaperScoreBean, String> keyedByPaperIdDS = javaBeanDS.keyBy(ExamPaperScoreBean::getPaper_id);
        //todo 7 开窗聚合
        SingleOutputStreamOperator<ExamPaperScoreBean> reduceDS = keyedByPaperIdDS.window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<ExamPaperScoreBean>() {
                    @Override
                    public ExamPaperScoreBean reduce(ExamPaperScoreBean value1, ExamPaperScoreBean value2) throws Exception {
                        value1.setACt(value1.getACt() + value2.getACt());
                        value1.setBCt(value1.getBCt() + value2.getBCt());
                        value1.setCCt(value1.getCCt() + value2.getCCt());
                        value1.setDCt(value1.getDCt() + value2.getDCt());
                        return value1;
                    }
                }, new WindowFunction<ExamPaperScoreBean, ExamPaperScoreBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<ExamPaperScoreBean> input, Collector<ExamPaperScoreBean> out) throws Exception {
                        ExamPaperScoreBean next = input.iterator().next();
                        out.collect(MyWindowFunction.getWindow(next, window));
                    }
                });

        //todo 8 关联维表，获取试卷名称
        SingleOutputStreamOperator<ExamPaperScoreBean> resultDS = AsyncDataStream.unorderedWait(reduceDS, new AsyncDimFunction<ExamPaperScoreBean>("DIM_TEST_PAPER") {
            @Override
            public void getDimInfo(ExamPaperScoreBean input, JSONObject dimInfo) {
                String paper_title = dimInfo.getString("PAPER_TITLE");
                input.setPaper_name(paper_title);
            }

            @Override
            public String getId(ExamPaperScoreBean input) {
                return input.getPaper_id();
            }

            @Override
            public String getFilterId() {
                return "id";
            }
        }, 60, TimeUnit.SECONDS);
        //todo 8 写入clickHouse
        resultDS.print("reduceDS>>>>>>>>>>>>>>>>>>");
        resultDS.addSink(ClickHouseUtil_hjy.getClickHouseSink("insert into dws_exam_paper_score_window values(?,?,?,?,?,?,?,?,?)"));
        //todo 9 启动程序
        env.execute();
    }
}
